package com.toutiao.vssue

import java.util.Properties

import kafka.javaapi.producer.Producer
import kafka.producer.{KeyedMessage, ProducerConfig}

class ProducerStore(
        brokers:String,
        topic:String
    ) {
  
  private val prop = new Properties
  
  prop.put("metadata.broker.list",brokers)
  prop.put("serializer.class", "kafka.serializer.StringEncoder")
  prop.put("request.required.acks","1")
  
  private val producer = new Producer[String,String](new ProducerConfig(prop))
  
  def send(tasks:Array[String]) = {
    tasks.par.foreach{task =>
      producer.send(new KeyedMessage[String,String](topic,task))
      }
  }
  
}

//import org.apache.log4j.Logger
//import org.apache.log4j.PropertyConfigurator
//
//object TestPp extends App {
//  
//  val log = Logger.getLogger(this.getClass.getName)
//  
//  PropertyConfigurator.configure("log4j.properties")  
//  
//  val br = "10.4.160.136:9095,10.4.160.137:9095,10.4.160.138:9095,10.4.160.139:9095,10.4.160.140:9095,10.4.161.43:9095,10.4.161.44:9095,10.4.161.45:9095,10.4.161.46:9095,10.4.161.47:9095,10.4.161.95:9095,10.4.161.96:9095,10.4.161.97:9095,10.4.161.98:9095,10.4.161.99:9095,10.4.161.163:9095,10.4.161.164:9095,10.4.161.165:9095,10.4.161.166:9095,10.4.161.167:9095"  
//  val tp = "crawl_article_list_sim_cluster_write"
//  
//  val p = new ProducerStore(tp,br)
//  
//  val xx = Array("xxx zzz")
//  p.send(xx)
//  
//  log.info("SEND DONE")
//  
//}
